RAG 技术深度解析

from langchain.document_loaders import (
    TextLoader,
    PyPDFLoader,
    DirectoryLoader,
    WebBaseLoader
)

# 加载单个文本文件
loader = TextLoader("doc.txt")
documents = loader.load()

# 加载 PDF
pdf_loader = PyPDFLoader("report.pdf")
pdf_docs = pdf_loader.load()

# 加载整个目录
dir_loader = DirectoryLoader(
    "./docs",
    glob="**/*.md",
    loader_cls=TextLoader
)
all_docs = dir_loader.load()

# 加载网页
web_loader = WebBaseLoader([
    "https://example.com/doc1",
    "https://example.com/doc2"
])
web_docs = web_loader.load()

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    TokenTextSplitter
)

# 字符分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每块大小
    chunk_overlap=200,    # 重叠大小
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]
)

chunks = text_splitter.split_documents(documents)

# Token 分块
token_splitter = TokenTextSplitter(
    chunk_size=500,       # token 数量
    chunk_overlap=50,
    encoding_name="cl100k_base"
)

token_chunks = token_splitter.split_documents(documents)

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

# OpenAI Embeddings
openai_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"
)

# 本地 Embeddings
local_embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# 生成向量
texts = [chunk.page_content for chunk in chunks]
vectors = openai_embeddings.embed_documents(texts)

from langchain.vectorstores import Chroma, Pinecone, FAISS
from langchain_community.vectorstores import Qdrant

# Chroma (本地，最简单)
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=openai_embeddings,
    persist_directory="./chroma_db"
)

# FAISS (本地，高效)
faiss_store = FAISS.from_documents(
    documents=chunks,
    embedding=openai_embeddings
)
faiss_store.save_local("faiss_index")

# Pinecone (云端)
pinecone_store = Pinecone.from_documents(
    documents=chunks,
    embedding=openai_embeddings,
    index_name="my-rag-index"
)

# Qdrant (本地/云端)
qdrant_store = Qdrant.from_documents(
    documents=chunks,
    embedding=openai_embeddings,
    url="http://localhost:6333",
    collection_name="my_collection"
)

# 相似度检索
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}  # 返回 Top-4
)

# 相似度得分检索
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 4,
        "score_threshold": 0.7  # 相似度阈值
    }
)

# MMR 检索（多样性）
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "lambda_mult": 0.5}
)

# 执行检索
results = retriever.get_relevant_documents("什么是 RAG？")

for i, doc in enumerate(results):
    print(f"文档 {i+1}:")
    print(f"内容: {doc.page_content[:100]}...")
    print(f"元数据: {doc.metadata}")
    print()

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

# 自定义 Prompt
prompt_template = """请基于以下文档回答问题。如果文档中没有相关信息，请说"我不知道"。

文档内容:
{context}

问题: {question}

回答:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# 创建 RAG 链
llm = ChatOpenAI(model="gpt-4o", temperature=0)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    chain_type_kwargs={"prompt": PROMPT}
)

# 查询
result = qa_chain.invoke({"query": "什么是 RAG？"})

print("回答:", result['result'])
print("\n来源文档:")
for doc in result['source_documents']:
    print(f"- {doc.metadata.get('source', 'unknown')}")

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 目标块大小
    chunk_overlap=200,      # 重叠大小（保持上下文）
    length_function=len,
    separators=[
        "\n\n",    # 段落
        "\n",      # 行
        "。",      # 中文句号
        "！",      # 中文感叹号
        "？",      # 中文问号
        "，",      # 中文逗号
        " ",       # 空格
        ""         # 字符（最后手段）
    ]
)

chunks = text_splitter.split_documents(documents)

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# 子文档（用于检索）
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400
)

# 父文档（用于生成）
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=2000
)

# 创建检索器
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=InMemoryStore(),
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

# 添加文档
retriever.add_documents(documents)

# 检索时返回父文档
results = retriever.get_relevant_documents("问题")

# 基础相似度检索
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

# 带阈值
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 5,
        "score_threshold": 0.7
    }
)

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,
        "lambda_mult": 0.5
    }
)

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# 向量检索
vector_retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

# 关键词检索 (BM25)
bm25_retriever = BM25Retriever.from_documents(
    documents=chunks
)
bm25_retriever.k = 5

# 混合检索
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.5, 0.5]  # 权重
)

from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 加载重排序模型
reranker = HuggingFaceCrossEncoder(
    model_name="BAAI/bge-reranker-base"
)

# 检索更多文档
docs = retriever.get_relevant_documents("问题", k=20)

# 重排序
reranked_docs = reranker.compress_documents(
    documents=docs,
    query="问题"
)

# 取 Top-K
final_docs = reranked_docs[:5]

def query_expansion(query: str, llm) -> list:
    """扩展查询为多个变体"""
    prompt = f"""生成 3 个不同方式表达的查询，意图与以下查询相同：
查询: {query}

只输出查询，每行一个。"""

    result = llm.invoke(prompt)
    queries = [query] + result.content.split('\n')
    return [q.strip() for q in queries if q.strip()]

# 使用扩展查询
queries = query_expansion("如何提高 RAG 效果？", llm)

all_docs = []
for q in queries:
    docs = retriever.get_relevant_documents(q, k=2)
    all_docs.extend(docs)

# 去重
unique_docs = list({d.page_content: d for d in all_docs}.values())

# 自定义检索器
def metadata_filter_retrieval(query: str, filters: dict):
    """结合元数据过滤的检索"""
    # 先按相似度检索
    docs = vectorstore.similarity_search(query, k=20)

    # 再按元数据过滤
    filtered_docs = []
    for doc in docs:
        match = True
        for key, value in filters.items():
            if doc.metadata.get(key) != value:
                match = False
                break
        if match:
            filtered_docs.append(doc)

    return filtered_docs[:5]

# 使用
results = metadata_filter_retrieval(
    "API 使用方法",
    filters={"category": "技术文档", "year": 2024}
)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

# 只返回与问题相关的内容
compressed_docs = compression_retriever.get_relevant_documents("问题")

# 1. 优化分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 更小的块
    chunk_overlap=100,
    separators=["\n\n", "\n", "。", "！", "？"]
)

# 2. 使用领域 Embeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5"  # 中文优化
)

# 3. 查询重写
expanded_queries = query_expansion(original_query, llm)

# 1. 增加检索数量
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 10}  # 增加 k 值
)

# 2. 使用父文档检索
parent_retriever = ParentDocumentRetriever(...)

# 3. 迭代检索
def iterative_retrieval(query, max_iterations=3):
    context = ""
    for i in range(max_iterations):
        docs = retriever.get_relevant_documents(query)
        new_info = "\n".join([d.page_content for d in docs])
        if new_info not in context:
            context += "\n" + new_info

        # 检查是否足够
        if len(context) > 3000:
            break

        # 生成子问题
        query = llm.invoke(f"基于: {context}\n生成下一个子问题")
    return context

# 1. 改进 Prompt
strict_prompt = """请严格基于以下文档回答问题。
如果文档中没有答案，请说"根据提供的文档，我无法回答这个问题"。

文档:
{context}

问题: {question}

回答:"""

# 2. 添加来源验证
def add_citations(response, source_docs):
    """在回答中添加引用"""
    citations = "\n\n来源:\n"
    for doc in source_docs:
        citations += f"- {doc.metadata.get('source', 'unknown')}\n"
    return response + citations

# 3. 使用 Reranker 过滤低质量文档
reranker = HuggingFaceCrossEncoder(...)

from ragas import evaluate
from datasets import Dataset

# 准备评估数据
test_data = {
    "question": ["什么是 RAG？", "如何优化 RAG？"],
    "answer": ["RAG 是...", "优化方法包括..."],
    "contexts": [
        ["文档1内容", "文档2内容"],
        ["文档3内容", "文档4内容"]
    ],
    "ground_truth": ["RAG 的正确答案", "优化方法的正确答案"]
}

dataset = Dataset.from_dict(test_data)

# 评估
result = evaluate(
    dataset=dataset,
    metrics=[
        "faithfulness",     # 忠实度
        "answer_relevancy", # 相关性
        "context_precision",# 上下文精确度
        "context_recall"    # 上下文召回率
    ]
)

print(result)